package com.btcode.mq;

import com.btcode.common.MyLog;
import com.btcode.common.PropertiesUnit;
import com.btcode.exception.MsgException;
import com.btcode.log.ILog;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.QueueViewMBean;

/**
 * 消息中间件，用于调度模块、收发模块、数采模块以及UI间的通讯 Created by Chance on 2017/11/1.
 */
public final class MqKit {

    private ILog log;
    private Connection connection = null;
    private Session session = null;
    private Session syncSession = null;
    private Map<String, MessageProducer> producerMap = new HashMap<>();
    private Map<String, MessageConsumer> consumerMap = new HashMap<>();
    private static String brokerURL = null;

    private ExecutorService pool = Executors.newFixedThreadPool(
        Runtime.getRuntime().availableProcessors() * 2);
    private boolean isMqShutdown = false;

    private volatile static MqKit instance;

    private MqKit() {
        if (brokerURL == null) {
            throw new MsgException("brokerURL不能为空，请通过MqKit.setBrokerURL设置");
        }
        log = MyLog.getInstance().getLogger(MqKit.class);

        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
        connectionFactory.setTrustAllPackages(true);
        try {
            connection = connectionFactory.createConnection();
            ((ActiveMQConnection) connection).setUseAsyncSend(true);//配置异步发送
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            syncSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//专门用于同步取消息的
            log.info("MQ连接初始化成功");
        }
        catch (JMSException e) {
            log.error("MQ初始化异常：", e);
        }
        onDestroy();
    }

    public synchronized static void setBrokerURL(String brokerURL) {
        if (MqKit.brokerURL != null) {
            return;
        }
        MqKit.brokerURL = brokerURL;
    }

    public static MqKit getInstance() {
        if (instance == null) {
            synchronized (MqKit.class) {
                if (instance == null) {
                    instance = new MqKit();
                }
            }
        }
        return instance;
    }

    public void productMsg(String queueName, Object msg, boolean persistent) {
        productMsg(queueName, msg, persistent, session);
    }

    public void productSyncMsg(String queueName, Object msg, boolean persistent) {
        productMsg(queueName, msg, persistent, syncSession);
    }

    private void productMsg(String queueName, Object msg, boolean persistent, Session se) {

        if (msg == null || !(msg instanceof Serializable)) {
            return;
        }

        Destination des;
        MessageProducer pro;
        try {
            if (!producerMap.containsKey(queueName)) {
                des = se.createQueue(queueName);
                pro = se.createProducer(des);
                pro.setDeliveryMode(
                    persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);

                if (!persistent) {
                    pro.setTimeToLive(60 * 1000 * 3);
                }

                producerMap.put(queueName, pro);
            }
            else {
                pro = producerMap.get(queueName);
            }
            if (isMqShutdown) {
                return;
            }
            pool.execute(() -> {
                try {
                    pro.send(se.createObjectMessage((Serializable) msg));
                }
                catch (JMSException e) {
                    log.error("MQ消息发送异常：", e);
                }
            });
        }
        catch (JMSException e) {
            log.error("MQ初始化发送异常：", e);
        }
    }

    /**
     * 获取一条消息
     *
     * @param queueName 队列名
     * @return 收到的消息
     */
    public Message receive(String queueName) {
        Destination des;
        MessageConsumer cons;
        try {
            if (!consumerMap.containsKey(queueName)) {
                des = syncSession.createQueue(queueName);
                cons = syncSession.createConsumer(des);
                consumerMap.put(queueName, cons);
            }
            else {
                cons = consumerMap.get(queueName);
            }
            return cons.receive();
        }
        catch (Exception e) {
            log.error("消费队列" + queueName + "数据异常", e);
            throw new MsgException("消费队列" + queueName + "数据异常", e);
        }
    }

    public MessageConsumer createSyncConsumer(String queueName) {
        return createConsumerFromSession(queueName, syncSession);
    }

    public MessageConsumer createConsumer(String queueName) {
        return createConsumerFromSession(queueName, session);
    }

    public MessageConsumer createConsumerFromSession(String queueName, Session se) {
        Destination des;
        MessageConsumer cons;

        try {
            des = se.createQueue(queueName);
            cons = se.createConsumer(des);
            return cons;
        }
        catch (Exception e) {
            log.error("创建消费者出错", e);
            return null;
        }
    }

    public void consumeMsg(String queueName, MsgListener msgListener) {
        Destination des;
        MessageConsumer cons;
        try {
            if (!consumerMap.containsKey(queueName)) {
                des = session.createQueue(queueName);
                cons = session.createConsumer(des);
                consumerMap.put(queueName, cons);
            }
            else {
                cons = consumerMap.get(queueName);
            }
            cons.setMessageListener(message -> {
                if (message == null) {
                    return;
                }
                if (isMqShutdown) {
                    return;
                }
                pool.execute(() -> {
                    try {
                        message.acknowledge();
                        msgListener.onReceive(message);
                    }
                    catch (JMSException e) {
                        log.error("MQ消息接收异常：", e);
                    }
                });
            });
        }
        catch (JMSException e) {
            log.error("MQ初始化接收异常：", e);
        }
    }

    private void onDestroy() {
        Runtime run = Runtime.getRuntime();
        run.addShutdownHook(new Thread() {
            @Override
            public void run() {
                try {
                    isMqShutdown = true;
                    sleep(3000);
                    session.close();
                    syncSession.close();
                    connection.close();
                    pool.shutdown();
                    log.info("MQ服务销毁完成");
                }
                catch (Exception e) {
                    log.error("MQ销毁异常：", e);
                }
            }
        });
    }


    //    private final String connectorPath = "service:jmx:rmi:///jndi/rmi://127.0.0.1:1093/jmxrmi";
    Map<String, QueueViewMBean> queueViewMBeanMap = new HashMap<>();

    public QueueViewMBean getQueueViewMBean(String queueName) {
        if (!queueViewMBeanMap.containsKey(queueName)) {
            initQueueViewMBeanMap();
        }
        return queueViewMBeanMap.get(queueName);
    }

    private void initQueueViewMBeanMap() {
        String connectorPath = PropertiesUnit.getValue("mqJmxURL");
        try (JMXConnector connector = JMXConnectorFactory.connect(
            new JMXServiceURL(connectorPath))) {
            connector.connect();
            MBeanServerConnection connection = connector.getMBeanServerConnection();
            ObjectName name = new ObjectName(
                "org.apache.activemq:brokerName=localhost,type=Broker");
            BrokerViewMBean mBean = MBeanServerInvocationHandler.newProxyInstance(connection, name,
                                                                                  BrokerViewMBean.class,
                                                                                  true);

            for (ObjectName queueName : mBean.getQueues()) {
                QueueViewMBean queueMBean = MBeanServerInvocationHandler.newProxyInstance(
                    connection, queueName, QueueViewMBean.class, true);
                queueMBean.setUseCache(false);
                if (!queueViewMBeanMap.containsKey(queueMBean.getName())) {
                    queueViewMBeanMap.put(queueMBean.getName(), queueMBean);
                }
            }
        }
        catch (Exception e) {
            log.error("初始化MQ管理服务错误", e);
        }
    }

    public interface MsgListener {

        //        public MsgListener() {
        //
        //        }

        //        private Object holdObj;
        //
        //        public MsgListener(Object obj) {
        //            holdObj = obj;
        //        }
        //
        //        public Object getHoldObj() {
        //            return holdObj;
        //        }

        void onReceive(Message message);

        //        final public void acknowledgeMsg(Message message) {
        //            try {
        //                message.acknowledge();
        //            } catch (JMSException e) {
        //                log.error("MQ消息确认异常：", e);
        //            }
        //        }
    }
}
